package org.dragonnova.iot.business.cloud.transport.adapter;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Date;

import org.dragonnova.iot.business.cloud.conf.Constants;
import org.dragonnova.iot.business.cloud.mq.message.TagMessage;
import org.dragonnova.iot.business.cloud.mq.service.KafkaService;
import org.dragonnova.iot.business.cloud.transport.model.AckPacket;
import org.dragonnova.iot.business.cloud.transport.model.DataCommPacket;
import org.dragonnova.iot.business.cloud.transport.model.LogoffPacket;
import org.dragonnova.iot.business.cloud.transport.model.Packet;
import org.dragonnova.iot.business.cloud.transport.model.RegisterPacket;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.dubbo.common.json.JSON;
import com.alibaba.dubbo.common.json.JSONObject;

/**
 * 
 * @description: 业务消息转包处理
 * @author songxy DateTime 2017年3月21日 下午1:36:58
 * @company winter
 * @email thinkdata@163.com
 * @version 1.0
 */
@Sharable
public final class TagDecoderAdapter extends ChannelInboundHandlerAdapter {

	@Autowired
	private KafkaService kafkService;

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		System.out.println("<<<<<<<<<<< server is active >>>>>>>>>>>>");
		super.channelActive(ctx);
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg)
			throws Exception {
		if (msg instanceof ByteBuf) {
			ByteBuf buffer = (ByteBuf) msg;
			byte header = buffer.readByte();
			byte mt = buffer.readByte();
			dispatcher(mt, buffer, ctx);
			byte[] end = new byte[2];
			buffer.readBytes(end);
		}
	}

	private void dispatcher(byte mt, ByteBuf buffer, ChannelHandlerContext ctx)
			throws Exception {
		Packet packet = null;
		switch (mt) {
		case AckPacket.ACK_TYPE_CODE_UNSUPPORTED_OPERATION:
			packet = new AckPacket();
			ctx.writeAndFlush(packet);
			break;

		case RegisterPacket.REGISTE_TYPE_CODE_UNSUPPORTED_OPERATION:
			RegisterPacket register = new RegisterPacket();
			register.decodePacket(mt, buffer);
			packet = new AckPacket(AckPacket.ACK_TYPE_CODE_REGISTER);
			ctx.writeAndFlush(packet);
			break;

		case LogoffPacket.LOGOFF_TYPE_CODE_UNSUPPORTED_OPERATION:
			LogoffPacket logoff = new LogoffPacket();
			logoff.decodePacket(mt, buffer);
			packet = new AckPacket(AckPacket.ACK_TYPE_CODE_LOGOFF);
			ctx.writeAndFlush(packet);
			break;

		case DataCommPacket.DATACOMM_TYPE_CODE_UNSUPPORTED_OPERATION:
			DataCommPacket data = new DataCommPacket();
			data.decodePacket(mt, buffer);
			dispose(data, ctx);
			break;
		}

	}

	public void dispose(DataCommPacket data, ChannelHandlerContext ctx)
			throws Exception {
		JSONObject json = (JSONObject) JSON.parse(new String(data.getData(),
				"utf-8"));
		TagMessage tagMsg = new TagMessage();
		tagMsg.setPhone(data.getPhone());
		tagMsg.setTemperature(json.getFloat("temperature", (float) 888.0));
		tagMsg.setWetness(json.getFloat("wetness", (float) 888.0));
		tagMsg.setPackageCount((int) data.getSize());
		tagMsg.setCreateTime(new Date());
		tagMsg.setId(Constants.idGenerator.generateId().toString());
		tagMsg.setBattery((float) data.getPl());

		super.channelRead(ctx, tagMsg);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		System.out.println("server exception: " + cause.getMessage());
		ctx.close();
	}
}
